package com.example.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @application:
 * @author: LiYuhang
 * @Date: 2021/6/26 14:55
 * @version: 0.1
 * @Description :
 * @Modified By:
 */
public interface Producer {
    public static final String SERIALIZER="org.apache.kafka.common.serialization.StringSerializer";
    /**
     * 处理消息
     * @param message
     */
    void process(String message);

    /**
     * 写数据
     * @param producer
     * @param topic
     * @param message
     */
    static RecordMetadata write(KafkaProducer<String, String> producer,
                      String topic,String message) throws InterruptedException, ExecutionException, TimeoutException {
        ProducerRecord<String,String> producerRecord = new
                ProducerRecord(topic,message);
        Future<RecordMetadata> send = producer.send(producerRecord);
        return send.get(3,TimeUnit.SECONDS);
    }


    /**
     * 配置信息
     * @param servers
     * @return
     */
    static Properties createConfig(String servers){
        Properties config=new Properties();
        config.put("bootstrap.servers",servers);
        config.put("acks","all");
        config.put("retries","2");
        config.put("batch.size","1000");
        config.put("linger.ms","1");
        config.put("key.serializer",SERIALIZER);
        config.put("value.serializer",SERIALIZER);
        return config;
    }

    /**
     * 获取默认配置
     * @return
     */
    static Properties getDefaultConfig(){
        return createConfig("47.100.82.240:9092");
    }




}
